package reactor;

import reactor.InMemoryDataSource.Book;
import reactor.core.publisher.Flux;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class R01_ReactorDemo {
    public static void main(String[] args) {
        List<Book> books1 = getMostExpensiveBooksByCategory();
        books1.forEach(System.out::println);

        System.out.printf("%n%n");

        List<Book> books2 = getMostExpensiveBooksByCategoryFunctional();
        books2.forEach(System.out::println);

        System.out.printf("%n%n");

        Flux<Book> pipeline = getMostExpensiveBooksByCategoryReactive(Flux.just(InMemoryDataSource.books));
        pipeline = pipeline.doOnNext(System.out::println);
        System.out.println("什么都不会发生，直到 pipeline 开始");
        pipeline.subscribe();
    }

    // 返回一个包含每种类别中最贵的书的列表, 非函数式编程
    public static List<Book> getMostExpensiveBooksByCategory() {
        Map<String, Book> map = new HashMap<>();
        for (Book book : InMemoryDataSource.books) {
            Book aBook = map.get(book.getCategory());
            if (aBook != null) {
                if (book.getPrice().compareTo(aBook.getPrice()) > 0) {
                    map.put(book.getCategory(), book);
                }
            } else {
                map.put(book.getCategory(), book);
            }
        }
        return (new ArrayList<>(map.values()));
    }

    // 返回一个包含每种类别中最贵的书的列表, 函数式编程
    public static List<Book> getMostExpensiveBooksByCategoryFunctional() {
        return Stream.of(InMemoryDataSource.books)
                .collect(Collectors.groupingBy(Book::getCategory))
                // .collect(Collectors.groupingByConcurrent(Book::getCategory)) // 通过修改一行代码就可以编程并行代码
                .entrySet()
                .stream()
                // .parallelStream() // 通过修改一行代码就可以编程并行代码
                .map(e -> e.getValue()
                        .stream()
                        // .parallelStream() // 通过修改一行代码就可以编程并行代码
                        .sorted(Comparator.comparing(Book::getPrice).reversed())
                        .findFirst()
                        .get()
                ).collect(Collectors.toList());
    }

    // 返回一个包含每种类别中最贵的书的列表, 响应式编程
    public static Flux<Book> getMostExpensiveBooksByCategoryReactive(Flux<Book> books) {
        return books.collectMultimap(Book::getCategory)
                .flatMapMany(m -> Flux.fromIterable(m.entrySet()))
                .flatMap(e -> Flux.fromIterable(e.getValue())
                        .sort(Comparator.comparing(Book::getPrice).reversed())
                        .next());
    }
}
